package leo.peersim.abcast;

import java.util.Collections;
import java.util.HashMap;
import java.util.Vector;

import leo.peersim.common.FSMOperationRequest;
import leo.peersim.common.IFSAutomaton;
import leo.peersim.common.LamportEventComparator;
import leo.peersim.paxos.IClientRequest;

import org.apache.log4j.Logger;

import peersim.config.Configuration;
import peersim.core.CommonState;
import peersim.core.Linkable;
import peersim.core.Node;
import peersim.edsim.EDProtocol;
import peersim.transport.Transport;

/**
 * Atomic multicast protocol with fail-stop failure model assumption.
 * 
 * @author leo
 * @version June 12, 2009
 */
public class ABCast implements Linkable, EDProtocol, IFSAutomaton {
	
	private Logger logger = Logger.getLogger(this.getClass().getName());
	
	private int transportID;
	
	private long timestamp;
	private long largestAgreedTimestamp;
	
	private Vector<Node> replicas = null;
	
	private HashMap<String, ABCastRound> managedTransactions;
	private Vector<ABCastRound> holdBackQueue;
	private Vector<FSMOperationRequest> deliveryQueue;
	
	private DebugStats DEBUG_stats;
	
	/**
	 * Reads the options from configuration files and sets protocols variables
	 * (global to all network's nodes). 
	 * 
	 * @param prefix
	 */
	public ABCast(String prefix) {
		this.transportID = Configuration.getPid(prefix + "." + "transport");
		
		this.DEBUG_stats = new DebugStats(); // Global to all the network!!
		this.init();
	}
	
	/**
	 * 
	 */
	private void init() {
		this.replicas = new Vector<Node>();
		
		this.timestamp = 0;
		this.largestAgreedTimestamp = 0;
		
		this.managedTransactions = new HashMap<String, ABCastRound>();
		this.holdBackQueue = new Vector<ABCastRound>();
		this.deliveryQueue = new Vector<FSMOperationRequest>();
	}
	
	/**
	 * Method called by the peersim engines to instantiate the protocol on each node of
	 * the network. It defines the protocol's variables which are local on each node.
	 */
	public ABCast clone() {
		ABCast abcast = null;
		try {
			abcast = (ABCast) super.clone();
		} catch (CloneNotSupportedException e) {
			logger.error("Error while cloning"); //It should never happens
		}
		
		abcast.init();
		
		logger.trace("Object cloned!");
		return abcast;
	}

	/**
	 * Handler for messages sender to this node from a remote agent. Since our
	 * protocol is made of a numbered of messages issued by agents which plays
	 * different roles, we map each agent to a specific event. This method acts
	 * as a dispatcher.
	 */
	@Override
	public void processEvent(Node localNode, int protocolID, Object event) {
		/*
		 * FIXME dispatching is done by relying on dynamic cast (instanceof). Is it
		 * possible to use double dispatching like in Visitor design pattern instead?
		 */
		if (event instanceof Request) {
			this.processMyEvent(localNode, protocolID, (Request) event);
		} else if (event instanceof Propose) {
			this.processMyEvent(localNode, protocolID, (Propose) event);
		} else if (event instanceof Agree) {
			this.processMyEvent(localNode, protocolID, (Agree) event);
		} else if (event instanceof FSMOperationRequest) {
			this.requestAction(localNode, protocolID, (FSMOperationRequest)event);
		} else {
			logger.error("Dispatching error!!");
		}
	}
	
	/**
	 * Handler for client request to perform an action onto the fault-tolerant
	 * Finite State Machine 
	 * 
	 * @param localNode Node from which the protocol is invoked
	 * @param protocolID identifier of the protocol on the node
	 */
	private void requestAction(Node localNode, int protocolID, FSMOperationRequest clientRequest) {
		this.timestamp++;
		
		String transactionID = "" + localNode.getID() + this.timestamp;
		
		ABCastRound t = new ABCastRound(transactionID, localNode, this.timestamp, localNode, this.timestamp, clientRequest);
		if ( this.managedTransactions.put(transactionID, t) != null) {
			logger.error("There is already a transaction with identifier _" + transactionID + "_");
			return;
		}
		
		this.holdBackQueue.add(t);
		
		Request reqMessage = new Request(transactionID, localNode, this.timestamp, clientRequest);
		
		this.broadcastMessage(localNode, protocolID, this.replicas, reqMessage);
		
		logger.debug(getNS(localNode) + "TID:" + t.getID() + " starting...");	
	}
	
	/**
	 * Handler for incoming Request
	 * 
	 * @param localNode
	 * @param protocolID
	 * @param event Request message as scheduled by the protocol.
	 */
	private void processMyEvent(Node localNode, int protocolID, Request event) {
		
		long receivedTimestamp = event.getLamportTimestamp();
		this.updateTimestampOnReceive(receivedTimestamp);
		
		String transactionID = event.getTransactionID();
		
		ABCastRound t = this.managedTransactions.get(transactionID);
		if (t!=null) {
			logger.error("Contains already TID:" + transactionID);
			//TODO
		} else {
			
			logger.debug(getNS(localNode) + "TID:" + transactionID + "[REQUEST] from agent[" + event.getSenderNode().getID() + "]");
			
			long proposedTimestamp = this.timestamp;
			FSMOperationRequest clientRequestLocalToEachReplica = event.getContent().clone();
			t = new ABCastRound(transactionID, event.getSenderNode(), event.getLamportTimestamp(), localNode, proposedTimestamp, clientRequestLocalToEachReplica);
			if ( this.managedTransactions.put(transactionID, t) != null) {
				logger.error("There is already a transaction with identifier _" + transactionID + "_");
				return;
			}
			
			this.holdBackQueue.add(t);
			
			this.timestamp++;
			ABCastMessage proposeMessage = new Propose(transactionID, localNode, this.timestamp, proposedTimestamp);
			Node receipt = event.getSenderNode();
			
			Transport trLayer = (Transport) receipt.getProtocol(transportID);
			trLayer.send(localNode, receipt, proposeMessage, protocolID);
			this.DEBUG_stats.addMessage(new NetworkMessage(trLayer.getLatency(localNode, receipt), proposeMessage.getSizeInByte()));
		}
	}
	
	/**
	 * Handler for incoming {@link Propose}.
	 * 
	 * @param localNode
	 * @param protocolID
	 * @param event
	 */
	private void processMyEvent(Node localNode, int protocolID, Propose event) {
		
		long receivedTimestamp = event.getLamportTimestamp();
		this.updateTimestampOnReceive(receivedTimestamp);
		
		String tid = event.getTransactionID();
		
		ABCastRound t = this.managedTransactions.get(tid);
		if (t==null) {
			logger.error("TID:" + tid + " unknown"); //TID stands for Transaction ID
			//TODO
		} else {
			
			Node proposer = event.getSenderNode();
			if (!t.addProposal(proposer, event.getProposedTimestamp())) {
				logger.error("I already have a proposal from agent " + proposer.getID());
				//TODO
			}
			
			logger.debug(getNS(localNode) + "TID:" + tid + "[PROPOSE] agent[" + event.getSenderNode().getID() + "] proposed ts:" + event.getProposedTimestamp() );
			
			if (t.getProposals().size() == this.replicas.size()) {
				
				HashMap<Node,Long> proposals = t.getProposals();
				this.largestAgreedTimestamp++;
				long agreedTimestamp = this.largestAgreedTimestamp;
				for (Long proposal : proposals.values()) {
					agreedTimestamp = Math.max(agreedTimestamp, proposal.longValue());
				}
				this.largestAgreedTimestamp = agreedTimestamp; //FIXED 20090613
				
				this.timestamp++; // Is it necessary?
				
				t.agree(agreedTimestamp);
				this.updateDeliveryQueue();
				
				logger.debug(getNS(localNode) + "TID:" + tid + " agreeded ts:" + agreedTimestamp + ", broadcasting to group...");
				
				ABCastMessage messageToSend = new Agree(tid, localNode, this.timestamp, agreedTimestamp);
				this.broadcastMessage(localNode, protocolID, this.replicas, messageToSend);
				logger.debug(getNS(localNode) + "delivery queue:" + this.deliveryQueue);
			}
		}		
	}
	
	/**
	 * Handler for incoming {@link Agreed}
	 * 
	 * @param localNode
	 * @param protocolID
	 * @param event
	 */
	protected void processMyEvent(Node localNode, int protocolID, Agree event) {
		
		long receivedTimestamp = event.getLamportTimestamp();
		this.updateTimestampOnReceive(receivedTimestamp);
		
		String tid = event.getTransactionID();
		
		ABCastRound t = this.managedTransactions.get(tid);
		if (t==null) {
			logger.error(getNS(localNode) + "TID:" + tid + " not known");
			//TODO
		} else {
			
			long agreedTimestamp = event.getAgreedTimestamp();
			
			logger.debug(getNS(localNode) + "TID:" + tid + "[AGREE] ts:" + agreedTimestamp);
			
			t.agree(agreedTimestamp);
			this.updateDeliveryQueue();
			
			logger.debug(getNS(localNode) + "delivery queue:" + this.deliveryQueue);
			
		}	
	}
	
	/**
	 * Broadcast to all agents within the group
	 * 
	 * @param localNode
	 * @param protocolID
	 * @param broadcastGroup
	 * @param message
	 * @return
	 */
	private Vector<Node> broadcastMessage(Node localNode, int protocolID, Vector<Node> broadcastGroup, Object message) {
		
		Vector<Node> recipients = new Vector<Node>();
		
		for (Node recipient : broadcastGroup) {
			
			recipients.add(recipient);
			
			Transport trLayer = (Transport) recipient.getProtocol(transportID);
			//logger.debug("Sending message to " + receipt.getID() + " (est. " + trLayer.getLatency(localNode, receipt) + " latency)");		
			trLayer.send(localNode, recipient, message, protocolID);
			
			this.DEBUG_stats.addMessage(new NetworkMessage(trLayer.getLatency(localNode, recipient), ((ABCastMessage)message).getSizeInByte()));

		}
		return recipients;
		
	}
	
	/**
	 * 
	 */
	private void updateDeliveryQueue() {
		
		if (!this.holdBackQueue.isEmpty()) {
		
			Collections.sort(this.holdBackQueue, new LamportEventComparator());
			while( this.holdBackQueue.firstElement().hasBeenCompleted() ) {
				
				FSMOperationRequest op = this.holdBackQueue.firstElement().getFSMCommand();
				op.setExecutedTime(CommonState.getTime());
				//logger.info(getNS(CommonState.getNode()) + "executing operation " + op.getOperazione() + "...");
				this.deliveryQueue.add( op );
				
				this.holdBackQueue.remove(0);
				
				if(this.holdBackQueue.isEmpty()) {
					break;
				}
			}
		}
	}
	
	/**
	 * Shortcut used for pretty logging
	 * @param node
	 * @return
	 */
	private String getNS(Node node) {
		return CommonState.getTime() + "\t[" + node.getID() + "]:" + this.timestamp + " ";
	}
	
	/**
	 * 
	 * @param receivedTimestamp
	 */
	private void updateTimestampOnReceive(long receivedTimestamp) {
		this.timestamp = Math.max(this.timestamp, receivedTimestamp);
		this.timestamp+=1;
	}
	
	public HashMap<String, ABCastRound> getManagedTransactions() {
		return this.managedTransactions;
	}
	
	public DebugStats getDebufInfo() {
		return this.DEBUG_stats;
	}
	
	public Vector<FSMOperationRequest> getDeliveryQueue() {
		return this.deliveryQueue;
	}
	
	//Following methods implements the Linkable interface

	@Override
	public boolean addNeighbor(Node arg0) {
		return replicas.add(arg0);
	}

	@Override
	public boolean contains(Node arg0) {
		throw new RuntimeException("Not yet supported");
	}

	@Override
	public int degree() {
		//logger.debug(" il gruppo ha una dimensione " + gruppo.size());
		return this.replicas.size();
	}

	@Override
	public Node getNeighbor(int index) {
		return replicas.get(index);
	}

	@Override
	public void pack() {
		throw new RuntimeException("Not yet supported");
	}

	@Override
	public void onKill() {
		throw new RuntimeException("Not yet supported");
	}

	/**
	 * FIXME encapsulation please re-factor
	 */
	@Override
	public Vector<String> getCommandHistory() {
		Vector<String> ret = new Vector<String>();
		
		for (FSMOperationRequest r : this.deliveryQueue) {
			ret.add(r.getOperazione());
		}
		return ret;
	}
	
	public Vector<IClientRequest> getClientRequests() {
		//TODO
		return null;
	}
	
}

/**
 * @author leo
 */
//TODO still insights
class NetworkMessage {
	
	private long messageID;
	private long sentTime;
	private long receivedTime;
	private long latency;
	
	private int sizeInBytes;
	
	private Transport tr;
	
	public NetworkMessage(Transport trasportProtocol) {
		this.tr = trasportProtocol;
	}
	
	//FIXME Dirty, please refactor me!
	public NetworkMessage(long latency, int size) {
		this.sentTime = CommonState.getTime(); //  because it's called only by the sender
		this.receivedTime = this.sentTime + latency;
		this.sizeInBytes = size;
	}
	
	public int getSizeInBytes() {
		return this.sizeInBytes;
	}
	
	public void send() {
		this.sentTime = CommonState.getTime();
		
		//arg0 sourceNode
		//arg1 destination
		//arg2 message
		//arg3 protocolID
		//this.tr.send(arg0, arg1, arg2, arg3)
		
	}
	
	public void receive() {
		this.receivedTime = CommonState.getTime();
	}
	
	public long getLatency() {
		return this.receivedTime - this.sentTime;
	}

	public long getSentTime() {
		return this.sentTime;
	}
	
}

@Deprecated
class DebugStats {
	
	private Vector<NetworkMessage> messagesSent;
	
	public DebugStats() {
		messagesSent = new Vector<NetworkMessage>();
	}
	
	public void addMessage(NetworkMessage message) {
		this.messagesSent.add(message);
	}
	
	public Vector<NetworkMessage> getLatenze() {
		//return latenze;
		return this.messagesSent;
	}
	
}

